package com.xxl.job.admin.core.thread;

import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.cron.CronExpression;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.scheduler.MisfireStrategyEnum;
import com.xxl.job.admin.core.scheduler.ScheduleTypeEnum;
import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author xuxueli 2019-05-21
 */
public class JobScheduleHelper {
    private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);

    private static JobScheduleHelper instance = new JobScheduleHelper();

    public static JobScheduleHelper getInstance() {
        return instance;
    }

    public static final long PRE_READ_MS = 5000;    // pre read

    private Thread scheduleThread;
    private Thread ringThread;
    private volatile boolean scheduleThreadToStop = false;
    private volatile boolean ringThreadToStop = false;
    private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();

    public void start() {

        // schedule thread
        scheduleThread = new Thread(new Runnable() {
            @Override
            public void run() {

                try {
                    /**
                     * 睡眠到5s
                     * System.currentTimeMillis() % 1000)为当前时间戳的毫秒位
                     *
                     * 5000 - System.currentTimeMillis() % 1000为需要睡的时间
                     *
                     * 例如System.currentTimeMillis() = 1734593483057 即 2024-12-19 15:31:23
                     * 则需要睡 5000-1734593483057 % 1000 = 5000 - 57 = 4943
                     *
                     * 即睡到 1734593488000 = 2024-12-19 15:31:28秒
                     *
                     **/
                    TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
                } catch (InterruptedException e) {
                    if (!scheduleThreadToStop) {
                        logger.error(e.getMessage(), e);
                    }
                }
                logger.info(">>>>>>>>> init xxl-job admin scheduler success.");

                // pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20)
                int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax() + XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;

                while (!scheduleThreadToStop) {

                    // Scan Job
                    long start = System.currentTimeMillis();

                    Connection conn = null;
                    Boolean connAutoCommit = null;
                    PreparedStatement preparedStatement = null;

                    boolean preReadSuc = true;
                    try {

                        conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
                        connAutoCommit = conn.getAutoCommit();
                        conn.setAutoCommit(false);

                        preparedStatement = conn.prepareStatement("select * from xxl_job_lock where lock_name = 'schedule_lock' for update");
                        preparedStatement.execute();

                        // tx start

                        // 1、pre read
                        long nowTime = System.currentTimeMillis();
                        List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
                        if (scheduleList != null && scheduleList.size() > 0) {
                            // 2、push time-ring
                            for (XxlJobInfo jobInfo : scheduleList) {

                                // time-ring jump
                                if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
                                    // 2.1、trigger-expire > 5s：pass && make next-trigger-time
                                    // 这种情况是triggerNextTime下次调度时间 + 5s比我当前时间还要小，说明这个任务已经错过了它的触发时间，有misfire误触发
                                    // 根据误触发策略（jobInfo.getMisfireStrategy()），决定如何处理这个任务。默认为什么都不做(MisfireStrategyEnum.DO_NOTHING)
                                    logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());

                                    // 1、misfire match
                                    MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum.match(jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
                                    if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
                                        // FIRE_ONCE_NOW 》 trigger
                                        // 如果我得策略是“立即触发一次”（FIRE_ONCE_NOW）
                                        JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null, null, null);
                                        logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId());
                                    }

                                    // 2、fresh next
                                    // 最后，无论采取何种策略，都会更新任务的下一个有效触发时间（refreshNextValidTime(jobInfo, new Date())）。
                                    refreshNextValidTime(jobInfo, new Date());

                                } else if (nowTime > jobInfo.getTriggerNextTime()) {
                                    // 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time

                                    /**
                                     * 如果nowTime大于jobInfo.getTriggerNextTime()，
                                     * 说明任务已经应该被触发了，但可能由于某种原因（如系统延迟、锁竞争等）还没有被实际触发。
                                     **/

                                    // 1、trigger
                                    // 触发任务
                                    JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null, null);
                                    logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId());

                                    // 2、fresh next
                                    refreshNextValidTime(jobInfo, new Date());

                                    // next-trigger-time in 5s, pre-read again
                                    /**
                                     * 如果任务的状态是启用的（jobInfo.getTriggerStatus() == 1），
                                     * 并且当前时间加上一个预读时间阈值（PRE_READ_MS）仍然大于任务的下一个触发时间，
                                     * 说明在下次触发之前，系统需要再次检查这个任务。
                                     **/
                                    if (jobInfo.getTriggerStatus() == 1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {

                                        // 1、make ring second
                                        /**
                                         * 计算任务下一个触发时间的分钟内的秒数（ringSecond），这可能是为了某种时间轮算法或调度策略做准备。
                                         *
                                         * 目的：计算下一次触发时间是在某个整数分钟后的几秒
                                         * 例如： 1734624012345L这个时间戳，翻译过来就是2024-12-20 00:00:12
                                         * 1734624012345L / 1000 = 1734624012
                                         * 1734624012 % 60 = 12
                                         * ringSecond = 12s
                                         *
                                         * 说明下一个触发时间是在12秒之后触发
                                         **/
                                        int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);

                                        // 2、push time ring
                                        pushTimeRing(ringSecond, jobInfo.getId());

                                        // 3、fresh next

                                        // 再次调用refreshNextValidTime方法，
                                        // 但这次传入的是任务原本的下一个触发时间new Date(jobInfo.getTriggerNextTime())作为计算下一个触发时间的起点
                                        // （这里可能有些冗余，因为在上一步已经刷新过，但可能是为了确保时间的一致性或更新某些内部状态）。
                                        refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

                                    }

                                } else {
                                    // 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time
                                    /**
                                     * nowTime < jobInfo.getTriggerNextTime()
                                     * 当前时间比下一个触发时间小，说明任务应该在某个时间点被触发，但当前时间还没有到，需要等待。
                                     **/

                                    // 1、make ring second
                                    int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);

                                    // 2、push time ring
                                    pushTimeRing(ringSecond, jobInfo.getId());

                                    // 3、fresh next
                                    refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

                                }

                            }

                            // 3、update trigger info
                            for (XxlJobInfo jobInfo : scheduleList) {
                                // 更新xxl_job_info表中的状态trigger_last_time、trigger_next_time、trigger_status
                                XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
                            }

                        } else {
                            preReadSuc = false;
                        }

                        // tx stop


                    } catch (Exception e) {
                        if (!scheduleThreadToStop) {
                            logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
                        }
                    } finally {

                        // commit
                        // 如果数据库连接（conn）不为null，则尝试提交当前事务。
                        // 如果在提交过程中发生SQLException异常，并且scheduleThreadToStop变量为false（意味着当前线程没有被计划停止），则记录错误日志。
                        if (conn != null) {
                            try {
                                conn.commit();
                            } catch (SQLException e) {
                                if (!scheduleThreadToStop) {
                                    logger.error(e.getMessage(), e);
                                }
                            }
                            // 尝试将数据库连接的自动提交模式恢复到之前的状态（connAutoCommit变量的值）。
                            // 这通常是在执行了一系列需要手动提交的事务操作之后进行的，以确保后续的数据库操作按照预期的自动提交模式进行。
                            try {
                                conn.setAutoCommit(connAutoCommit);
                            } catch (SQLException e) {
                                if (!scheduleThreadToStop) {
                                    logger.error(e.getMessage(), e);
                                }
                            }
                            // 尝试关闭数据库连接。如果关闭过程中发生异常，并且线程没有被计划停止，则记录错误日志。
                            try {
                                conn.close();
                            } catch (SQLException e) {
                                if (!scheduleThreadToStop) {
                                    logger.error(e.getMessage(), e);
                                }
                            }
                        }

                        // close PreparedStatement
                        // 如果预处理语句（preparedStatement）不为null，则尝试关闭它。
                        // 预处理语句用于执行参数化的SQL语句，关闭它可以释放数据库资源。
                        // 如果关闭过程中发生异常，并且线程没有被计划停止，则记录错误日志。
                        if (null != preparedStatement) {
                            try {
                                preparedStatement.close();
                            } catch (SQLException e) {
                                if (!scheduleThreadToStop) {
                                    logger.error(e.getMessage(), e);
                                }
                            }
                        }
                    }
                    long cost = System.currentTimeMillis() - start;


                    // Wait seconds, align second
                    if (cost < 1000) {  // scan-overtime, not wait
                        try {
                            // pre-read period: success > scan each second; fail > skip this period;
                            // job预读成功则用1000，否则用PRE_READ_MS（5000）减去当前时间的毫秒数
                            TimeUnit.MILLISECONDS.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
                        } catch (InterruptedException e) {
                            if (!scheduleThreadToStop) {
                                logger.error(e.getMessage(), e);
                            }
                        }
                    }

                }

                logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
            }
        });
        scheduleThread.setDaemon(true);
        scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
        scheduleThread.start();


        // ring thread
        ringThread = new Thread(new Runnable() {
            @Override
            public void run() {

                while (!ringThreadToStop) {

                    // align second
                    try {
                        TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
                    } catch (InterruptedException e) {
                        if (!ringThreadToStop) {
                            logger.error(e.getMessage(), e);
                        }
                    }

                    try {
                        // second data
                        List<Integer> ringItemData = new ArrayList<>();
                        // 获取当前秒数0-59
                        int nowSecond = Calendar.getInstance().get(Calendar.SECOND);   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
                        for (int i = 0; i < 2; i++) {
                            // 在代码 List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60); 中，
                            // 使用 +60 的目的是为了确保当 nowSecond 是一个较小的值（比如接近0）时，
                            // 通过减去 i（其中 i 是从0开始的循环变量）不会得到一个负数索引。
                            // 在Java中，列表的索引不能是负数，否则会抛出 IndexOutOfBoundsException。
                            List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);
                            if (tmpData != null) {
                                ringItemData.addAll(tmpData);
                            }
                        }

                        // ring trigger
                        logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData));
                        if (ringItemData.size() > 0) {
                            // do trigger
                            for (int jobId : ringItemData) {
                                // do trigger
                                JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
                            }
                            // clear
                            ringItemData.clear();
                        }
                    } catch (Exception e) {
                        if (!ringThreadToStop) {
                            logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
                        }
                    }
                }
                logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
            }
        });
        ringThread.setDaemon(true);
        ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread");
        ringThread.start();
    }

    /**
     * XXL-JOB任务调度平台中用于刷新任务下一次有效触发时间的逻辑。
     *
     * @param jobInfo  XxlJobInfo类型的对象，包含了任务的详细信息，如任务ID、触发状态、上次触发时间、下次触发时间等。
     * @param fromTime Date类型的对象，表示计算下一次有效触发时间的起始时间点。
     * @throws Exception
     */
    private void refreshNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
        try {
            Date nextValidTime = generateNextValidTime(jobInfo, fromTime);
            if (nextValidTime != null) {
                jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
                jobInfo.setTriggerNextTime(nextValidTime.getTime());
            } else {
                // generateNextValidTime fail, stop job
                jobInfo.setTriggerStatus(0);
                jobInfo.setTriggerLastTime(0);
                jobInfo.setTriggerNextTime(0);
                logger.error(">>>>>>>>>>> xxl-job, refreshNextValidTime fail for job: jobId={}, scheduleType={}, scheduleConf={}",
                        jobInfo.getId(), jobInfo.getScheduleType(), jobInfo.getScheduleConf());
            }
        } catch (Exception e) {
            // generateNextValidTime error, stop job
            jobInfo.setTriggerStatus(0);
            jobInfo.setTriggerLastTime(0);
            jobInfo.setTriggerNextTime(0);

            logger.error(">>>>>>>>>>> xxl-job, refreshNextValidTime error for job: jobId={}, scheduleType={}, scheduleConf={}",
                    jobInfo.getId(), jobInfo.getScheduleType(), jobInfo.getScheduleConf(), e);
        }
    }

    /**
     * 将job添加到时间轮
     * 时间轮理论见博客：https://blog.csdn.net/weixin_45683778/article/details/144576957
     *
     * @param ringSecond ringSecond（一个整数，表示分钟内的秒数）
     * @param jobId      jobId（一个整数，表示任务的ID）
     */
    private void pushTimeRing(int ringSecond, int jobId) {
        // push async ring
        List<Integer> ringItemData = ringData.get(ringSecond);
        if (ringItemData == null) {
            ringItemData = new ArrayList<Integer>();
            ringData.put(ringSecond, ringItemData);
        }
        ringItemData.add(jobId);

        logger.debug(">>>>>>>>>>> xxl-job, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData));
    }

    public void toStop() {

        // 1、stop schedule
        scheduleThreadToStop = true;
        try {
            TimeUnit.SECONDS.sleep(1);  // wait
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
        if (scheduleThread.getState() != Thread.State.TERMINATED) {
            // interrupt and wait
            scheduleThread.interrupt();
            try {
                scheduleThread.join();
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }

        // if has ring data
        boolean hasRingData = false;
        if (!ringData.isEmpty()) {
            for (int second : ringData.keySet()) {
                List<Integer> tmpData = ringData.get(second);
                if (tmpData != null && tmpData.size() > 0) {
                    hasRingData = true;
                    break;
                }
            }
        }
        if (hasRingData) {
            try {
                TimeUnit.SECONDS.sleep(8);
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }

        // stop ring (wait job-in-memory stop)
        ringThreadToStop = true;
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
        if (ringThread.getState() != Thread.State.TERMINATED) {
            // interrupt and wait
            ringThread.interrupt();
            try {
                ringThread.join();
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }

        logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop");
    }


    /**
     * 根据任务的调度类型和配置来计算下一次有效触发时间。
     *
     * @param jobInfo  XxlJobInfo类型的对象，包含了任务的调度类型scheduleType和调度配置scheduleConf等信息。
     * @param fromTime Date类型的对象，表示计算下一次有效触发时间的起始时间点。
     * @return Date类型，表示计算出的下一次有效触发时间；如果无法计算出下一次触发时间，则返回null。
     * @throws Exception 可能会抛出异常，例如解析调度配置时出错等。
     */
    // ---------------------- tools ----------------------
    public static Date generateNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
        ScheduleTypeEnum scheduleTypeEnum = ScheduleTypeEnum.match(jobInfo.getScheduleType(), null);
        if (ScheduleTypeEnum.CRON == scheduleTypeEnum) {
            Date nextValidTime = new CronExpression(jobInfo.getScheduleConf()).getNextValidTimeAfter(fromTime);
            return nextValidTime;
        } else if (ScheduleTypeEnum.FIX_RATE == scheduleTypeEnum /*|| ScheduleTypeEnum.FIX_DELAY == scheduleTypeEnum*/) {
            // 如果调度类型是FIX_RATE（或者注释中提到的FIX_DELAY，但当前代码只处理了FIX_RATE），
            // 则直接将fromTime的时间加上调度配置中指定的毫秒数（Integer.valueOf(jobInfo.getScheduleConf())*1000），
            // 得到下一次触发时间。这里假设jobInfo.getScheduleConf()返回的是一个以毫秒为单位的整数字符串。
            return new Date(fromTime.getTime() + Integer.valueOf(jobInfo.getScheduleConf()) * 1000);
        }
        return null;
    }

    /**
     * 测试CronExpression计算
     **/
    public static void main(String[] args) {

        // corn表达式 5分钟执行一次，计算当前时间的下一次有效触发时间
        // CRON表达式 "0 0/5 * * * ?" 的含义是：
        //  每5分钟执行一次。
        //  0 秒：在每分钟的第0秒触发。
        //  0/5 分：从第0分钟开始，每隔5分钟触发一次（即0分、5分、10分、15分等）。
        //  * 小时：每个小时。
        //  * 日：每天。
        //  * 月：每个月。
        //  ? 星期：不指定。

        String cron = "0 0/5 * * * ?";
        try {
            Date nowDate = new Date();
            Date nextValidTime = new CronExpression(cron).getNextValidTimeAfter(nowDate);
            Date nextNextValidTime = new CronExpression(cron).getNextValidTimeAfter(nextValidTime);
            System.out.println("Now time: " + nowDate);
            System.out.println("Next valid time: " + nextValidTime);
            System.out.println("Next Next valid time: " + nextNextValidTime);
            // Now time: Thu Dec 19 11:36:10 CST 2024
            // Next valid time: Thu Dec 19 11:40:00 CST 2024
            // Next Next valid time: Thu Dec 19 11:45:00 CST 2024
            // 第一次结果取当前时间最近的整数倍，当前时间为36分，最近的5分钟是40分，所以第一次结果是40分。
            // 第二次结果取第一次结果之后的整数倍，当前时间为40分，最近的5分钟是45分，所以第二次结果是45分。

            // 计算轮数
            int ringSecond = (int) ((1734624012345L / 1000) % 60);
            System.out.println("ringSecond: " + ringSecond);

            // 测试时间轮
            JobScheduleHelper jobScheduleHelper = new JobScheduleHelper();
            for (int i = 0; i < 3; i++){
                // 生成随机时间戳
                long time = System.currentTimeMillis();
                int ringSec = (int) ((time / 1000) % 60);
                Thread.sleep(1000);
                jobScheduleHelper.pushTimeRing(ringSec, i);
            }
            System.out.println(ringData);
            System.out.println(5000 - System.currentTimeMillis() % 1000);

            int nowSecond = Calendar.getInstance().get(Calendar.SECOND);
            int nowMiSecond = Calendar.getInstance().get(Calendar.MILLISECOND);
            System.out.println("nowSecond: " + nowSecond);
            System.out.println("nowMiSecond: " + nowMiSecond);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
